package com.aiwiown.snackmq.broker.transaction;

import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageType;
import com.aiwiown.snackmq.network.connection.ConnectionManager;
import com.aiwiown.snackmq.storage.transaction.TransactionLog;
import com.aiwiown.snackmq.storage.transaction.TransactionRecord;
import io.netty.channel.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.Collections;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 事务状态回查服务。
 * 定期扫描事务日志，对超时的半消息向生产者发起状态回查。
 */
@Slf4j
@RequiredArgsConstructor
public class TransactionCheckerService {

    private final TransactionLog transactionLog;
    private final ConnectionManager connectionManager;
    private final long transactionTimeoutMs; // 事务超时时间
    private final long checkIntervalMs;      // 回查任务执行间隔

    private final ScheduledExecutorService checkerScheduler =
            Executors.newSingleThreadScheduledExecutor(r -> new Thread(r, "TransactionCheckerThread"));

    public void start() {
        checkerScheduler.scheduleAtFixedRate(this::check, checkIntervalMs, checkIntervalMs, TimeUnit.MILLISECONDS);
        log.info("TransactionCheckerService started. Check interval: {}ms, transaction timeout: {}ms.",
                checkIntervalMs, transactionTimeoutMs);
    }

    public void shutdown() {
        checkerScheduler.shutdown();
    }

    private void check() {
        try {
            Map<String, TransactionRecord> pendingTransactions = transactionLog.getPendingTransactions();
            if (pendingTransactions.isEmpty()) {
                return;
            }

            log.debug("Checking {} pending transactions...", pendingTransactions.size());
            long now = System.currentTimeMillis();

            for (TransactionRecord record : pendingTransactions.values()) {
                if (now - record.getTimestamp() > transactionTimeoutMs) {
                    log.warn("Transaction {} has timed out. Initiating check-back.", record.getTransactionId());
                    Channel producerChannel = connectionManager.getChannelByProducerId(record.getProducerId());

                    if (producerChannel != null && producerChannel.isActive()) {
                        // 构建回查请求
                        // 将原始半消息作为 body 发送回去，方便客户端识别
                        Message checkRequest = Message.builder()
                                .type(MessageType.CHECK_TRANSACTION_STATE)
                                .property("transactionId", record.getTransactionId())
                                .messages(Collections.singletonList(record.getMessage()))
                                .build();

                        producerChannel.writeAndFlush(checkRequest).addListener(future -> {
                            if (!future.isSuccess()) {
                                log.error("Failed to send transaction check request for transactionId: {}",
                                        record.getTransactionId(), future.cause());
                            }
                        });
                    } else {
                        log.warn("Cannot check transaction {}: producer '{}' is not connected.",
                                record.getTransactionId(), record.getProducerId());
                    }
                }
            }
        } catch (Exception e) {
            log.error("Error during transaction check.", e);
        }
    }
}